package com.coastline.fruit.mq.service.impl;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.CharsetUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.coastline.fruit.mq.entity.FailMsgEntity;
import com.coastline.fruit.mq.exception.MsgException;
import com.coastline.fruit.mq.service.FailMsgService;
import com.coastline.fruit.mq.service.MQService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.retry.RetryState;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * 发送消息实现类
 *
 * @author Coastline
 * @date 2025/2/19 17:38
 */
@Slf4j
@Service
// 只有当配置中的前缀为caostline的名为mq 的值是true 的时候才会加载该组件
@ConditionalOnProperty(prefix = "coastline", value = "mq", havingValue = "true")
public class MQServiceImpl implements MQService {
    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private FailMsgService failMsgService;

    /**
     * 发送即时消息
     * 设置自动重试机制，自动重试三次，第一次重试时间是三秒，以后的每次重试是上一次重试时间的1.5倍
     *
     * @param exchange   交换机
     * @param routingKey 路由键
     * @param msg        消息
     * @return boolean 结果
     */
    @Override
    @Retryable(value = MsgException.class, maxAttempts = 2, backoff = @Backoff(value = 3000, multiplier = 1.5))
    public boolean sendMsg(String exchange, String routingKey, Object msg) {
        return this.sendMsg(exchange, routingKey, msg, -1);
    }

    /**
     * 发送延迟消息
     *
     * @param exchange   交换机
     * @param routingKey 路由键
     * @param msg        消息
     * @param delay      延迟时间
     * @return boolean 结果
     */
    @Override
    @Retryable(value = MsgException.class, maxAttempts = 2, backoff = @Backoff(value = 3000, multiplier = 1.5))
    public boolean sendMsg(String exchange, String routingKey, Object msg, int delay) {
        // 检测并转换消息格式
        String jsonMessage = (msg instanceof String) ? (String) msg : JSONUtil.toJsonStr(msg);

        // 获取消息绑定的唯一id
        String fastUUID = IdUtil.fastUUID();
        log.info("消息发送成功，消息ID：{}，交换机：{}，路由键：{}，消息：{}", fastUUID, exchange, routingKey, jsonMessage);
        CorrelationData correlationData = new CorrelationData(fastUUID);

        correlationData.getFuture().addCallback(result ->
                {
                    if (result.isAck()) {
                        // 消息发送成功
                        log.debug("消息发送成功, ID: {}", correlationData.getId());
                    } else {
                        log.error("消息发送失败：ID: {}, 原因：{}", correlationData.getId(), correlationData.getReturned());
                        FailMsgEntity failMsgEntity = FailMsgEntity.builder()
                                .exchange(exchange)
                                .msgId(correlationData.getId())
                                .routingKey(routingKey) // 根据情况可选择不存储
                                .msg(jsonMessage)
                                .reason(result.getReason()).build();
                        this.failMsgService.save(failMsgEntity);
                    }
                },
                ex -> {
                    String exceptionInfo = ExceptionUtil.getMessage(ex);
                    log.error("消息发送失败, message id ：{}, reason：{}", correlationData.getId(), exceptionInfo);
                    FailMsgEntity failMsgEntity = FailMsgEntity.builder()
                            .exchange(exchange)
                            .msgId(correlationData.getId())
                            .routingKey(routingKey) // 根据情况可选择不存储
                            .msg(jsonMessage)
                            .reason(exceptionInfo).build();
                    this.failMsgService.save(failMsgEntity);
                });
        Message message = MessageBuilder.withBody(StrUtil.bytes(jsonMessage, CharsetUtil.CHARSET_UTF_8))
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();

        try {
            /**
             * messagePostProcessor 消息后处理器
             * 作用：1、用于在消息发送前对消息做修改 2、在这里是通过其中的setDelay设置当前消息延迟时间（单位为毫秒）
             */
            rabbitTemplate.convertAndSend(exchange, routingKey, message, messagePostProcessor ->{
                messagePostProcessor.getMessageProperties().setDelay(delay);
                return messagePostProcessor;
            });
        } catch (Exception e){
            throw MsgException.builder()
                    .failMsgEntity(FailMsgEntity.builder().exchange(exchange)
                            .msgId(correlationData.getId()).msg(jsonMessage)
                            .reason(ExceptionUtil.getMessage(e)).build()).build();
        }
        return true;
    }
}
